package rocketmq

import (
	"context"
	"errors"
	"gitee.com/vrv_media/go-micro-framework/pkg/mq"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/apache/rocketmq-client-go/v2/producer"
	"sync"
)

// rocketMQ 基于rocketMq的consumer和producer的统一实现
type rocketMQ struct {
	addr              []string
	retry             int //  重试次数
	consumerGroup     string
	consumerClient    rocketmq.PushConsumer
	subscriber        *SubscribeHandler
	consumerTopic     string
	producerGroup     string
	broadCastConsumer bool                   // 是否是消费广播消息,默认false
	queueSelector     producer.QueueSelector // 发送消息queue的选择器，主要用于顺序发送消息
	producerClient    rocketmq.Producer
	once              sync.Once
}

// NewConsumer 创建集群消息消费端
func NewConsumer(addr []string, consumerGroup string, consumerTopic string, subscriber *SubscribeHandler) (mq.Consumer, error) {
	return newCon(addr, consumerGroup, false, consumerTopic, subscriber)
}

// NewBroadCastConsumer 创建广播消息消费端
func NewBroadCastConsumer(addr []string, consumerGroup string, consumerTopic string, subscriber *SubscribeHandler) (mq.Consumer, error) {
	return newCon(addr, consumerGroup, true, consumerTopic, subscriber)
}

func newCon(addr []string, consumerGroup string, broadCastConsumer bool, consumerTopic string, subscriber *SubscribeHandler) (mq.Consumer, error) {
	if len(addr) == 0 {
		return nil, errors.New("empty addr")
	}
	if consumerGroup == "" {
		return nil, errors.New("empty ConsumerGroup")
	}
	if subscriber == nil {
		return nil, errors.New("subscriber handler is nil")
	}
	if consumerTopic == "" {
		return nil, errors.New("empty consumer topic")
	}
	if subscriber.Func == nil {
		return nil, errors.New("empty consumer func")
	}
	return &rocketMQ{
		addr:              addr,
		consumerGroup:     consumerGroup,
		subscriber:        subscriber,
		broadCastConsumer: broadCastConsumer,
		consumerTopic:     consumerTopic,
		retry:             3,
	}, nil
}
func (r *rocketMQ) WithRetry(retry int) *rocketMQ {
	r.retry = retry
	return r
}

func (r *rocketMQ) ConsumerStart(opts ...consumer.Option) error {
	var err error
	r.once.Do(func() {
		var consumerOpts []consumer.Option
		if len(r.addr) > 0 {
			consumerOpts = append(consumerOpts, consumer.WithNsResolver(
				primitive.NewPassthroughResolver(r.addr),
			))
		} else {
			err = errors.New("rocketmq Addr not found")
			return
		}
		if r.consumerGroup != "" {
			consumerOpts = append(consumerOpts, consumer.WithGroupName(r.consumerGroup))
		}

		if r.retry > 0 {
			consumerOpts = append(consumerOpts, consumer.WithRetry(r.retry))
		}
		if r.broadCastConsumer {
			consumerOpts = append(consumerOpts, consumer.WithConsumerModel(consumer.BroadCasting))
		} else {
			consumerOpts = append(consumerOpts, consumer.WithConsumerModel(consumer.Clustering))
		}
		if len(opts) > 0 {
			consumerOpts = append(consumerOpts, opts...)
		}
		r.consumerClient, err = rocketmq.NewPushConsumer(consumerOpts...)
		if err != nil {
			return
		}
		err = r.Subscribe()
		if err != nil {
			return
		}
		if err = r.consumerClient.Start(); err != nil {
			return
		}
	})
	return err
}

func (r *rocketMQ) Subscribe() error {
	err := r.consumerClient.Subscribe(
		r.consumerTopic,
		r.subscriber.Selector,
		r.subscriber.Func,
	)
	if err != nil {
		return err
	}
	return nil
}

func (r *rocketMQ) ConsumerStop() error {
	if r.consumerClient != nil {
		return r.consumerClient.Shutdown()
	}
	return nil
}

func NewProducer(addr []string, producerGroup string, selector producer.QueueSelector) (mq.Producer, error) {
	return NewProducerWithRetry(addr, producerGroup, 3, selector)
}

func NewProducerWithRetry(addr []string, producerGroup string, retry int, selector producer.QueueSelector) (mq.Producer, error) {
	if len(addr) == 0 {
		return nil, errors.New("empty addr")
	}
	return &rocketMQ{
		addr:          addr,
		retry:         retry,
		producerGroup: producerGroup,
		queueSelector: selector,
	}, nil
}

func (r *rocketMQ) ProducerStart() error {
	var err error

	r.once.Do(func() {
		var producerOpts []producer.Option
		if len(r.addr) > 0 {
			producerOpts = append(producerOpts, producer.WithNsResolver(
				primitive.NewPassthroughResolver(r.addr),
			))
		} else {
			err = errors.New("rocketmq Addr not found")
			return
		}

		if r.producerGroup != "" {
			producerOpts = append(producerOpts, producer.WithGroupName(r.producerGroup))
		}
		if r.queueSelector != nil {
			producerOpts = append(producerOpts, producer.WithQueueSelector(r.queueSelector))
		}
		if r.retry > 0 {
			producerOpts = append(producerOpts, producer.WithRetry(r.retry))
		}

		r.producerClient, err = rocketmq.NewProducer(producerOpts...)
		if err != nil {
			return
		}
		if err = r.producerClient.Start(); err != nil {
			return
		}

	})
	return err
}

func (r *rocketMQ) ProducerStop() error {
	if r.producerClient != nil {
		return r.producerClient.Shutdown()
	}
	return nil
}

func (r *rocketMQ) MessageSync(topic string, data []byte, shardingKeyValue *string) error {
	msg := primitive.NewMessage(topic, data)
	if shardingKeyValue != nil && *shardingKeyValue != "" {
		msg.WithShardingKey(*shardingKeyValue)
	}
	_, err := r.producerClient.SendSync(context.Background(), msg)
	return err
}

// MessageSyncDelayTime set message delay time to consume.
// reference delay level definition: 1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
// delay level starts from 1. for example, if we set param level=1, then the delay time is 1s.
func (r *rocketMQ) MessageSyncDelayTime(topic string, data []byte, delayTimeLevel int, shardingKeyValue *string) error {
	msg := primitive.NewMessage(topic, data)
	msg.WithDelayTimeLevel(delayTimeLevel)
	if shardingKeyValue != nil && *shardingKeyValue != "" {
		msg.WithShardingKey(*shardingKeyValue)
	}
	_, err := r.producerClient.SendSync(context.Background(), msg)
	return err
}

func (r *rocketMQ) MessageAsync(topic string, data []byte,
	callback func(ctx context.Context, result *primitive.SendResult, err error),
	shardingKeyValue *string) error {
	msg := primitive.NewMessage(topic, data)
	if shardingKeyValue != nil && *shardingKeyValue != "" {
		msg.WithShardingKey(*shardingKeyValue)
	}
	return r.producerClient.SendAsync(context.Background(), callback, msg)
}

func (r *rocketMQ) MessageOneWay(topic string, data []byte, shardingKeyValue *string) error {
	msg := primitive.NewMessage(topic, data)
	if shardingKeyValue != nil && *shardingKeyValue != "" {
		msg.WithShardingKey(*shardingKeyValue)
	}
	return r.producerClient.SendOneWay(context.Background(), msg)
}

type SubscribeHandler struct {
	// 对 消息的属性进行条件过滤，例如标签、键值对 等
	Selector consumer.MessageSelector
	// 订阅完消息的回调函数
	Func func(context.Context, ...*primitive.MessageExt) (consumer.ConsumeResult, error)
}
